#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "main.h"
#include "shmem.h"
#include "sqlite3.h"

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define CLIENTID_SEND "ExampleClientPub"
#define TOPIC_CMD "/app/data/down"
#define TOPIC_DATA "/app/data/up"
#define QOS 1
#define TIMEOUT 10000L
#define STM32 1
#define MODBUS 0

volatile MQTTClient_deliveryToken deliveredtoken;
pthread_mutex_t resource;
int report_type = 2;
int report_period = 3;
int send_flag = 1;

// sql数据库线程
void *sql_thread(void)
{
    while (1)
    {
        sqlite3 *db;
        char sql[128] = {0};
        time_t t = time(NULL);

        // 从共享内存取数据
        struct std_node *ptr = get_data_from_shm();
        // 将data存入
        for (int i = 0; i < get_num_from_shm(); i++)
        {
            // 判断类型
            if (ptr->type == 1) // bool
            {
                sprintf(sql, "insert into senor_data values(%d,%d,%d)",  ptr->key, ptr->new_val.b_val,t);
            }
            else if (ptr->type == 2) // int
            {
                sprintf(sql, "insert into senor_data values(%d,%d,%d)", ptr->key, ptr->new_val.i_val,t);
            }
            else if (ptr->type == 3) // float
            {
                sprintf(sql, "insert into senor_data values(%d,%f,%d)", ptr->key, ptr->new_val.f_val,t);
            }
            // 打开数据库
            if (sqlite3_open("/home/hq/Desktop/Database/database.db", &db) != SQLITE_OK)
            {
                fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
                return -1;
            }
            char *errmsg = NULL;
            int k = 0;
            sqlite3_exec(db, sql, NULL, NULL, &errmsg);
            ptr++;
        }
        printf("数据存储成功\n");
        sleep(60);
    }
}

// 模式修改Json格式处理
void mode_change_json(cJSON *root)
{
    cJSON *data = cJSON_GetObjectItem(root, "data");
    cJSON *type = cJSON_GetObjectItem(data, "type");
    cJSON *period = cJSON_GetObjectItem(data, "period");
    report_type = type->valueint;
    printf("模式已修改为%d\n",report_type); //测试
    report_period = period->valueint;
    printf("周期已修改为%d\n",report_period); //测试
}

// 从共享内存采集数据
struct std_node *get_data_from_shm(void)
{
    struct shm_param stm_senor;
    shm_init(&stm_senor, "data", 1024);
    int *num = (int *)shm_getaddr(&stm_senor);
    struct std_node *ptr = (struct std_node *)(num + 1);
    return ptr;
}

// 获取共享内存数据条数
int get_num_from_shm(void)
{
    struct shm_param stm_senor;
    shm_init(&stm_senor, "data", 1024);
    int *num = (int *)shm_getaddr(&stm_senor);
    return *num;
}

// 采集回复线程0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
void *send_data(void)
{
    printf("send_data\n");
    send_flag = 1;
}

// 控制指令线程
void *control_cmd(cJSON *root)
{
    pthread_mutex_lock(&resource);
    printf("control_cmd\n");
    printf("root:%s\n",cJSON_Print(root));
    // 打开消息队列
    Cmd cmd, cmd_test;
    cmd.type = 1;
    cJSON *data = cJSON_GetObjectItem(root, "data");
    cJSON *key = cJSON_GetObjectItem(data, "key");
    cJSON *val = cJSON_GetObjectItem(data, "val");
    // 判断是给stm32还是modbus
    // 打开共享内存
    struct shm_param para;
    shm_init(&para, "data", 1024);
    int *num = (int *)shm_getaddr(&para);
    int dev_type = -1;
    struct std_node *ptr = (struct std_node *)(num + 1);
    for (int i = 0; i < *num; i++)
    {
        if (ptr->key == key->valueint)
        {
            dev_type = ptr->dev_type;

            break;
        }
        ptr++;
    }
    if (dev_type == STM32)
    {
        // 序列化
        cJSON *cmd_js = cJSON_CreateObject();
        cJSON_AddNumberToObject(cmd_js, "key", key->valueint);
        cJSON_AddNumberToObject(cmd_js, "val", val->valueint);
        strcpy(cmd.commend, cJSON_Print(cmd_js));
        msg_queue_send("stm_command", (void *)(&cmd), sizeof(cmd), 0);
        printf("cmd:%s\n",cmd.commend);
        printf("发给stm32\n");
    }
    else if (dev_type == MODBUS)
    {
        // 序列化
        cJSON *cmd_js = cJSON_CreateObject();
        cJSON_AddNumberToObject(cmd_js, "key", key->valueint);
        cJSON_AddNumberToObject(cmd_js, "val", val->valueint);
        strcpy(cmd.commend, cJSON_Print(cmd_js));
        msg_queue_send("mb_command", (void *)(&cmd), sizeof(cmd), 0);
        printf("发送给modbus\n");
    }
    else
    {
        printf("dev_type error\n");
    }
    // 测试
    printf("cmd_test.commend:%s\n", cmd_test.commend);
    printf("control_cmd success\n");
    pthread_mutex_unlock(&resource);
}

// 模式修改线程
void *mode_change(cJSON *root)
{
    pthread_mutex_lock(&resource);
    printf("mode_change\n");
    // 解析Json
    mode_change_json(root);
    pthread_mutex_unlock(&resource);
}

// 自动模式修改线程
void *auto_mode_change(cJSON *root)
{
    pthread_mutex_lock(&resource);
    printf("auto_mode_change\n");
    // 解析Json
    cJSON *val = cJSON_GetObjectItem(root, "val");
    //编辑消息
    Cmd cmd ;
    cmd.type = 1;
    // 序列化
    cJSON *cmd_js = cJSON_CreateObject();
    cJSON_AddNumberToObject(cmd_js, "key", 501);
    cJSON_AddNumberToObject(cmd_js, "val", val->valueint);
    strcpy(cmd.commend, cJSON_Print(cmd_js));
    msg_queue_send("stm_command", (void *)(&cmd), sizeof(cmd), 0);
    printf("cmd:%s\n",cmd.commend);
    pthread_mutex_unlock(&resource);
}

// 历史消息请求线程
void *history_msg(cJSON *root)
{
    pthread_mutex_lock(&resource);
    printf("history_msg\n");
    // 解析JSON格式
    cJSON *data = cJSON_GetObjectItem(root, "data");
    // 获得key值
    cJSON *key = cJSON_GetObjectItem(data, "key");
    cJSON *limit = cJSON_GetObjectItem(data, "limit");
    int min = cJSON_GetArrayItem(limit, 0)->valueint;
    int max = cJSON_GetArrayItem(limit, 1)->valueint;
    printf("min:%d,max:%d\n", min, max); // 测试
    //创建发送的JSON格式
    cJSON *root_send = cJSON_CreateObject();
    cJSON_AddItemToObject(root_send,"type",cJSON_CreateNumber(4));
    cJSON_AddItemToObject(root_send,"result",cJSON_CreateNumber(0));
    cJSON_AddItemToObject(root_send,"key",cJSON_CreateNumber(key->valueint));
    cJSON *data_send = cJSON_CreateArray();
    //打开数据库
    sqlite3 *db;
    if (sqlite3_open("/home/hq/Desktop/Database/database.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    char *errmsg = NULL;
    // 从数据库查询数据
    char sql[128] = {0};
    printf("key:%d\n",key->valueint);
    sprintf(sql, "select * from senor_data where key = %d and time between %d and %d", key->valueint, min, max);
    //使用查询语句
    if(sqlite3_exec(db, sql, callback, (void*)data_send, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "select err:%s\n", errmsg);
    }
    else
    {
        printf("查询成功\n");
    }
    //将数组存入root_send
    cJSON_AddItemToObject(root_send,"data",data_send);
    //发送数据
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;
    // 定义MQTT消息
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    // 定义MQTT消息令牌
    MQTTClient_deliveryToken token;
    // 发布MQTT消息
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    // 创建MQTT客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID_SEND,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20; // 设置客户端与服务器保持连接的心跳间隔为 20 秒。
    conn_opts.cleansession = 1;       // 设置清除会话标志为 1，表示每次连接时都创建新会话。
    // 连接MQTT客户端
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 发送给客户端
    pubmsg.payloadlen = (int)strlen(cJSON_Print(root_send));
    pubmsg.payload = cJSON_Print(root_send);
    printf("发送给客户端:%s\n",pubmsg.payload);
    MQTTClient_publishMessage(client, TOPIC_DATA, &pubmsg, &token);
    pthread_mutex_unlock(&resource);
}
//查询回调
int callback(void *arg, int num, char **value, char **name)
{
    cJSON *data = (cJSON *)arg;
    cJSON *item = cJSON_CreateObject();
    printf("value[1]:%s,value[2]:%s\n",value[1],value[2]);
    cJSON_AddItemToObject(item,"time",cJSON_CreateNumber(atoi(value[2])));
    cJSON_AddItemToObject(item,"val",cJSON_CreateString(value[1]));
    cJSON_AddItemToArray(data,item);
    return 0;
}


void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    // 解析Json
    cJSON *root = cJSON_Parse(message->payload);
    cJSON *type = cJSON_GetObjectItem(root, "type");
    // 采集回复
    if (type->valueint == 1)
    {
        // 创建采集回复线程
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)send_data, NULL);
        pthread_detach(pid);
    }
    // 控制指令
    if (type->valueint == 2)
    {
        // 创建控制指令线程
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)control_cmd, root);
        pthread_detach(pid);
    }
    // 模式修改
    if (type->valueint == 3)
    {
        // 创建模式修改线程
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)mode_change, root);
        pthread_detach(pid);
    }
    // 历史消息请求
    if (type->valueint == 4)
    {
        // 创建历史消息请求线程
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)history_msg, root);
        pthread_detach(pid);
    }
    if(type->valueint == 5)
    {
        //创建修改模式(手动自动)
        pthread_t pid;
        pthread_create(&pid, NULL, (void *)auto_mode_change, root);
        pthread_detach(pid);
    }
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// 解析点表
void parse_point_table()
{
    // 打开文件(点表)
    FILE *fp = fopen("/mnt/config/node.json", "r");
    if (fp == NULL)
    {
        printf("open file failed\n");
        return;
    }
    // 读取文件内容
    char buf[2048];
    fread(buf, 1, 2048, fp);
    cJSON *root = cJSON_Parse(buf);
    // 解析定时上报周期数
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *period = cJSON_GetObjectItem(report, "period");
    report_period = period->valueint;
    // 解析stm32节点个数
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *stm_data = cJSON_GetObjectItem(stm32, "data");
    int data_num = cJSON_GetArraySize(stm_data);
    // 解析modbus节点个数
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    cJSON *mb_data = cJSON_GetObjectItem(modbus, "data");
    int modbus_num = cJSON_GetArraySize(mb_data);
    // 创建节点数组
    struct std_node node[data_num + modbus_num];
    // 解析STM32节点并赋值数组
    for (int i = 0; i < data_num; i++)
    {
        cJSON *data_i = cJSON_GetArrayItem(stm_data, i);
        cJSON *key = cJSON_GetObjectItem(data_i, "key");
        cJSON *type = cJSON_GetObjectItem(data_i, "type");
        node[i].key = key->valueint;
        node[i].type = type->valueint;
        node[i].dev_type = STM32;
        node[i].new_val.i_val = 0;
        node[i].old_val.i_val = 0;
        node[i].ret = -1;
    }
    // 解析modbus节点并赋值数组
    for (int i = 0; i < modbus_num; i++)
    {
        cJSON *modbus_i = cJSON_GetArrayItem(mb_data, i);
        cJSON *key = cJSON_GetObjectItem(modbus_i, "key");
        cJSON *type = cJSON_GetObjectItem(modbus_i, "type");
        node[i + data_num].key = key->valueint;
        node[i + data_num].type = type->valueint;
        node[i + data_num].dev_type = MODBUS;
        node[i + data_num].ret = -1;
        node[i + data_num].new_val.i_val = 0;
        node[i + data_num].old_val.i_val = 0;
    }
    // 将数组放入共享内存
    // 创建共享内存
    struct shm_param para;
    shm_init(&para, "data", 1024);
    // 将节点个数放入共享内存
    int *num = (int *)shm_getaddr(&para);
    *num = data_num + modbus_num;
    // 将节点数组放入共享内存
    struct std_node *node_shm = (struct std_node *)(num + 1);
    memcpy(node_shm, node, sizeof(node) * (data_num + modbus_num));
    // 遍历测试
    for (int i = 0; i < data_num + modbus_num; i++)
    {
        printf("node_shm[%d].key:%d\n", i, node_shm[i].key);
    }
}

int main(int argc, char *argv[])
{
    // 解析点表
    parse_point_table();
    pthread_mutex_init(&resource, NULL);
    // 开启数据库存储线程
    pthread_t pid;
    pthread_create(&pid, NULL, (void *)sql_thread, NULL);
    pthread_detach(pid);
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;
    // 定义MQTT消息
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    // 定义MQTT消息令牌
    MQTTClient_deliveryToken token;
    // 发布MQTT消息
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    // 创建MQTT客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20; // 设置客户端与服务器保持连接的心跳间隔为 20 秒。
    conn_opts.cleansession = 1;       // 设置清除会话标志为 1，表示每次连接时都创建新会话。
    // 设置MQTT客户端回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    // 连接MQTT客户端
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅MQTT主题
    MQTTClient_subscribe(client, TOPIC_CMD, QOS);

    // 等待用户输入
    while (1)
    {
        // 向QT客户端发送数据(定时发送)
        while (report_type == 2 && send_flag == 1)
        {
            sleep(report_period);
            if (report_type != 2)
            {
                break;
            }
            struct std_node *ptr = get_data_from_shm();
            // 创建JSON格式数据
            cJSON *root = cJSON_CreateObject();
            cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
            cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
            cJSON *data = cJSON_CreateArray();
            // 将data存入
            for (int i = 0; i < get_num_from_shm(); i++)
            {
                cJSON *data_i = cJSON_CreateObject();
                cJSON_AddItemToObject(data_i, "key", cJSON_CreateNumber(ptr->key));
                // 判断类型
                if (ptr->type == 1) // bool
                {
                    char value[10] = {0};
                    sprintf(value, "%d", ptr->new_val.b_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                else if (ptr->type == 2) // int
                {
                    char value[10] = {0};
                    sprintf(value, "%d", ptr->new_val.i_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                else if (ptr->type == 3) // float
                {
                    char value[10] = {0};
                    sprintf(value, "%f", ptr->new_val.f_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                cJSON_AddItemToArray(data, data_i);
                ptr++;
            }
            cJSON_AddItemToObject(root, "data", data);
            char *senor_data = cJSON_Print(root);
            printf("senor_data:%s\n",senor_data);
            // 发送给客户端
            pubmsg.payloadlen = (int)strlen(senor_data);
            pubmsg.payload = senor_data;
            MQTTClient_publishMessage(client, TOPIC_DATA, &pubmsg, &token);
        }
        // 向QT客户端发送数据(变化发送)
        while (report_type == 1 && send_flag == 1)
        {
            sleep(1);
            struct std_node *ptr = get_data_from_shm();
            // 创建JSON格式数据
            cJSON *root = cJSON_CreateObject();
            cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
            cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
            cJSON *data = cJSON_CreateArray();
            // 将data存入
            for (int i = 0; i < get_num_from_shm(); i++)
            {
                int flag = 0; // 默认无变化
                // 1-bool 2-int 3-float
                if (ptr->type == 1)
                {
                    if (ptr->new_val.b_val != ptr->old_val.b_val)
                    {
                        flag = 1;
                    }
                }
                else if (ptr->type == 2)
                {
                    if (ptr->new_val.i_val != ptr->old_val.i_val)
                    {
                        flag = 1;
                    }
                }
                else if (ptr->type == 3)
                {
                    if (ptr->new_val.f_val != ptr->old_val.f_val)
                    {
                        flag = 1;
                    }
                }
                if (flag != 1)
                {
                    break;
                }
                cJSON *data_i = cJSON_CreateObject();
                cJSON_AddItemToObject(data_i, "key", cJSON_CreateNumber(ptr->key));

                // 判断类型
                if (ptr->type == 1) // bool
                {
                    char value[10] = {0};
                    sprintf(value, "%d\0", ptr->new_val.b_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                else if (ptr->type == 2) // int
                {
                    char value[10] = {0};
                    sprintf(value, "%d", ptr->new_val.i_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                else if (ptr->type == 3) // float
                {
                    char value[10] = {0};
                    sprintf(value, "%f", ptr->new_val.f_val);
                    cJSON_AddItemToObject(data_i, "val", cJSON_CreateString(value));
                }
                cJSON_AddItemToArray(data, data_i);
                ptr++;
            }
            cJSON_AddItemToObject(root, "data", data);
            char *senor_data = cJSON_Print(root);
            printf("senor_data:%s\n",senor_data);//测试
            if (cJSON_GetArraySize(data) != 0)
            {
                // 发送给客户端
                pubmsg.payloadlen = (int)strlen(senor_data);
                pubmsg.payload = senor_data;
                MQTTClient_publishMessage(client, TOPIC_DATA, &pubmsg, &token);
            }
        }
    }
    // 取消订阅MQTT主题
    MQTTClient_unsubscribe(client, TOPIC_CMD);
    // 断开MQTT连接
    MQTTClient_disconnect(client, 10000);
    // 销毁MQTT客户端
    MQTTClient_destroy(&client);
    return rc;
}
